package com.example.mq.mqserver.datacenter;

import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.Binding;
import com.example.mq.mqserver.core.Exchange;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;
import lombok.extern.slf4j.Slf4j;

import java.beans.BeanInfo;
import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 管理内存中的数据
 * 由于会在多线程的场景下使用,需要注意线程安全
 */
@Slf4j
public class MemoryDataCenter {
    //在内存中需要管理 交换机、队列、绑定、消息

    //key: exchangeName, value: exchange对象
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    //key: queueName, value: queue对象
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    //key: exchangeName, value: 该交换机与队列的所有绑定关系映射集合(HashMap)
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingMap = new ConcurrentHashMap<>();
    //key: messageId, value: message对象
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    //key: queueName, value: 该队列下的消息对象的集合
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    //key: queueName, value: 该队列下未应答的消息映射集合
    //第二个key: messageId(UUID表示)
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

    //交换机
    public void insertExchange(Exchange exchange){
        exchangeMap.put(exchange.getName(), exchange);
        log.info("[MemoryDataCenter] 交换机插入成功, exchangeName:" + exchange.getName());
    }

    public void deleteExchange(String exchangeName){
        exchangeMap.remove(exchangeName);
        log.info("[MemoryDataCenter] 交换机删除成功, exchangeName:" + exchangeName);
    }

    public Exchange getExchange(String exchangeName){
        return exchangeMap.get(exchangeName);
    }

    //队列
    public void insertQueue(MSGQueue queue){
        queueMap.put(queue.getName(), queue);
        log.info("[MemoryDataCenter] 队列插入成功, queueName:" + queue.getName());
    }

    public void deleteQueue(String queueName){
        queueMap.remove(queueName);
        log.info("[MemoryDataCenter] 队列删除成功, queueName:" + queueName);
    }

    public MSGQueue getQueue(String queueName){
        return queueMap.get(queueName);
    }

    //绑定
    public void insertBinding(Binding binding) throws MqException {
//      ConcurrentHashMap<String, Binding> queueBindingMap = bindingMap.get(binding.getExchangeName());
        ConcurrentHashMap<String, Binding> queueBindingMap = bindingMap.computeIfAbsent(binding.getExchangeName(), k -> new ConcurrentHashMap<>());
        //检查嵌套的哈希表是否存在
        synchronized (queueBindingMap){
            //这样写需要再创建一把锁
//            if(queueBindingMap == null){
//                queueBindingMap = new ConcurrentHashMap<>();
//                bindingMap.put(binding.getExchangeName(), queueBindingMap);
//            }
            if(queueBindingMap.get(binding.getQueueName()) != null){
                throw new MqException("[MemoryDataCenter] 绑定关系已经存在 exchangeName:" + binding.getExchangeName() +
                        ", queueName" + binding.getQueueName());
            }
            queueBindingMap.put(binding.getQueueName(), binding);
            log.info("[MemoryDataCenter] 绑定关系插入成功 exchangeName:"+ binding.getExchangeName() +
                    ", queueName" + binding.getQueueName());
        }
    }

    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> queueBindingMap = bindingMap.get(binding.getExchangeName());
        if(queueBindingMap == null){
            throw new MqException("[MemoryDataCenter] 该绑定不存在 exchangeName:" + binding.getExchangeName() +
                    ", queueName" + binding.getQueueName());
        }
        bindingMap.remove(binding.getExchangeName());
        log.info("[MemoryDataCenter] 绑定关系删除成功 exchangeName:"+ binding.getExchangeName() +
                ", queueName" + binding.getQueueName());
    }

    //两个版本的获取绑定:
    //1.获取队列和交换机之间的唯一绑定
    public Binding getBinding(String exchangeName, String queueName) throws MqException {
        ConcurrentHashMap<String, Binding> queueBindingMap = bindingMap.get(exchangeName);
        if(queueBindingMap == null){
            return null;
        }
        return queueBindingMap.get(queueName);
    }
    //2.获取当前当前交换机下的所有绑定
    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName){
        return bindingMap.get(exchangeName);
    }

    //消息
    //向消息中心添加消息
    public void addMessage(Message message){
        messageMap.put(message.getMessageId(), message);
        log.info("[MemoryDataCenter] 消息添加完成 messageId:" + message.getMessageId());
    }

    public void removeMessage(String messageId){
        messageMap.remove(messageId);
        log.info("[MemoryDataCenter] 消息删除完成 messageId:" + messageId);
    }

    public Message getMessage(String messageId){
        return messageMap.get(messageId);
    }

    //将消息发送到指定队列
    public void sendMessage(String queueName, Message message){
//        LinkedList<Message> messages = queueMessageMap.get(queueName);
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queueName, k -> new LinkedList<>());
        synchronized (messages){
            //queueMessageMap.put(queueName, messages);
            //将消息添加到队列中
            messages.add(message);
        }
        //将消息添加到消息中心
        addMessage(message);
        log.info("[MemoryDataCenter] 消息已成功添加到队列 queueName:" + queueName + ", messageId:" + message.getMessageId());
    }

    //从指定队列中取消息
    public Message pollMessage(String queueName){
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        //当链表中没有元素
        if(messages == null){
            return null;
        }
       synchronized (messages){
           if(messages.size() == 0){
               return null;
           }
           Message message = messages.remove(0);
           log.info("[MemoryDataCenter] 消息已成功从队列中取出 queueName:" + queueName + ", messageId:" + message.getMessageId());
           return message;
       }
    }

    //获取指定队列中消息的个数
    public int getMessageSize(String queueName){
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages == null){
            return 0;
        }
        //LinkedList 线程不安全
        synchronized (messages){
            return messages.size();
        }
    }

    //添加未被确认消息
    public void addMessageWaitAck(String queueName, Message message){
//        ConcurrentHashMap<String, Message> messageWaitAck = queueMessageWaitAckMap.get(queueName);
        ConcurrentHashMap<String, Message> messageWaitAck = queueMessageWaitAckMap.computeIfAbsent(queueName, k -> new ConcurrentHashMap<>());
        synchronized (messageWaitAck){
//            if(messageWaitAck == null){
//                messageWaitAck = new ConcurrentHashMap<>();
//                queueMessageWaitAckMap.put(queueName, messageWaitAck);
//            }
            messageWaitAck.put(message.getMessageId(), message);
        }
        log.info("[MemoryDataCenter] 消息进入待确认队列 queueName:" + queueName + ", messageId:" + message.getMessageId());
    }

    //删除未确认消息(用户确认了消息)
    public void deleteMessageWaitAck(String queueName, String messageId){
        ConcurrentHashMap<String, Message> messageWaitAck = queueMessageWaitAckMap.get(queueName);
        if(messageWaitAck == null){
            return;
        }
        messageWaitAck.remove(messageId);
        log.info("[MemoryDataCenter] 消息已从待确认队列中删除 queueName:" + queueName + ", messageId:" + messageId);
    }

    //从待确认队列中获取指定消息
    public Message getMessageWaitAck(String queueName, String messageId){
        ConcurrentHashMap<String, Message> messageWaitAck = queueMessageWaitAckMap.get(queueName);
        if(messageWaitAck == null){
            return null;
        }
        return messageWaitAck.get(messageId);
    }

    //从硬盘上恢复数据到内存中
    public void recovery(DiskDataCenter diskDataCenter) throws MqException, IOException, ClassNotFoundException {
        //1.清空之前的数据
        exchangeMap.clear();
        queueMap.clear();
        bindingMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        //2.恢复交换机
        List<Exchange> exchangeList = diskDataCenter.selectAllExchanges();
        for(Exchange exchange : exchangeList){
            insertExchange(exchange);
        }
        //3.恢复队列
        List<MSGQueue> queueList = diskDataCenter.selectAllQueues();
        for(MSGQueue queue : queueList){
            insertQueue(queue);
        }
        //4.恢复绑定
        List<Binding> bindingList = diskDataCenter.selectAllBindings();
        for(Binding binding : bindingList){
            insertBinding(binding);
        }
        //5.恢复消息
        for(MSGQueue queue : queueList){
            LinkedList<Message> messageList = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            queueMessageMap.put(queue.getName(), messageList);
            for(Message message : messageList){
                addMessage(message);
            }
        }
    }
}
